package azalea.mqtt.gui;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import azalea.mqtt.gui.stream.MqttInputStream;
import azalea.mqtt.message.builder.ConnectionMsgBuilder;
import azalea.mqtt.message.builder.PingMsgBuilder;
import azalea.mqtt.message.builder.PublishMsgBuilder;
import azalea.mqtt.message.builder.SubscribeMsgBuilder;
import azalea.mqtt.message.builder.Topic;
import azalea.mqtt.parse.MqttData;
import azalea.mqtt.parse.MqttParser;
import azalea.mqtt.parse.data.CmdType;
import azalea.mqtt.parse.data.PubRecMqttData;
import azalea.mqtt.parse.data.PubRelMqttData;
import azalea.mqtt.parse.data.PublishMqttData;

public class MqttClient implements Runnable {
	Socket socket = new Socket();

	Thread readTh = new Thread(this);
	ExecutorService executorService;
	AtomicInteger id;
	AtomicBoolean connected = new AtomicBoolean(false);
	MqttClientPanel father;
	//InputStream _inputStream = null;
	MqttInputStream mqttInputStream=null;
	OutputStream outputStream = null;
	AtomicBoolean dorecv = new AtomicBoolean(false);
	//FileWriter fw;
	int keepLive = 0;
	int qos = 0;
	volatile long lastRcvTime = 0;

	public MqttClient(MqttClientPanel f) {
		/*try {
			fw = new FileWriter(new File("d:\\mqtt111.txt"));
		} catch (IOException e1) {
			e1.printStackTrace();
		}*/
		try {
			socket.setSoTimeout(5000);
		} catch (SocketException e) {
			e.printStackTrace();
		}
		executorService = Executors.newSingleThreadExecutor();
		father = f;
		Random r = new Random(System.currentTimeMillis());
		id = new AtomicInteger(r.nextInt(100));
	}

	private void sendBytes(byte[] sendMsg) throws IOException {
		// System.out.print("send：\r\n");
		// print(sendMsg, sendMsg.length);

		executorService.submit(new Runnable() {
			@Override
			public void run() {
				try {
					if (outputStream == null) {
						outputStream = socket.getOutputStream();
					}
					outputStream.write(sendMsg);
				} catch (Exception e) {
					e.printStackTrace();
					father.printInfo("错误：" + e.getMessage());
				}
			}
		});

	}


	public void disconnect() throws IOException {
		// TODO Auto-generated method stub
		/*fw.flush();
		fw.close();*/
		connected.set(false);
		dorecv.set(false);
		byte[] sendMsg = ConnectionMsgBuilder.disconnect();
		sendBytes(sendMsg);
		// readTh.stop();
		if (socket != null) {
			socket.close();
		}
		if (executorService != null) {
			executorService.shutdown();
		}
		father.printInfo("断开连接");
		father.connectFail();
	}

	public void connect(String host, String port, String clientId, String username, String password)
			throws IOException {
		dorecv.set(true);
		father.printInfo("准备连接：" + host);
		keepLive = father.getKeepLive();
		qos=father.getQos();
		int iport = Integer.parseInt(port);
		socket.connect(new InetSocketAddress(host, iport));
		readTh.start();
		byte[] sendMsg = ConnectionMsgBuilder.connectV3_1(true, true, false, qos, false, true, keepLive, clientId, null, null,
				username, password);
		sendBytes(sendMsg);
		executorService.submit(new Runnable() {

			@Override
			public void run() {
				// TODO Auto-generated method stub
				try {
					InputStream inputStream = socket.getInputStream();
					byte[] readbuf = new byte[10240];
					int length = inputStream.read(readbuf);

					MqttData mqttData = MqttParser.parse(readbuf,length);
					father.updateRcvCommand(mqttData.getCmdType().name());
					if (mqttData.getCmdType() != CmdType.CONNACK) {
						throw new RuntimeException("连接失败");
					}
					father.printInfo("连接成功");
					father.connectSuccess();
					updateRcvTime();
					connected.set(true);
					mqttInputStream =new MqttInputStream(inputStream);
				} catch (Exception e) {
					e.printStackTrace();
					father.printInfo("错误：" + e.getMessage());
					connected.set(false);
				}
			}
		});

	}

	public void ping() throws IOException {
		byte[] sendMsg = PingMsgBuilder.pingReq();
		sendBytes(sendMsg);
	}

	public void pong() throws IOException {
		byte[] sendMsg = PingMsgBuilder.pingResp();
		sendBytes(sendMsg);
	}

	public void puback(int id) throws IOException {
		byte[] sendMsg = PublishMsgBuilder.puback(id);
		sendBytes(sendMsg);
	}
	public void pubrel(int id) throws IOException {
		byte[] sendMsg = PublishMsgBuilder.pubrel(id);
		sendBytes(sendMsg);
	}
	public void pubrec(int id) throws IOException{
		byte[] sendMsg = PublishMsgBuilder.pubrec(id);
		sendBytes(sendMsg);
	}
	public void pubcomp(int id) throws IOException{
		byte[] sendMsg = PublishMsgBuilder.pubcomp(id);
		sendBytes(sendMsg);
	}
	public void subscribe(String text) throws IOException {
		// TODO Auto-generated method stub
		int identifier = id.addAndGet(1);
		Topic topic = new Topic(text, 0);
		byte[] sendMsg = SubscribeMsgBuilder.subscribe(identifier, topic);
		sendBytes(sendMsg);
	}

	public void dissubscribe(String text) throws IOException {
		// TODO Auto-generated method stub
		int identifier = id.addAndGet(1);
		byte[] sendMsg = SubscribeMsgBuilder.unsubscribe(identifier, text);
		sendBytes(sendMsg);
	}

	public void sendTxt(String topic, String txt) throws IOException {
		// TODO Auto-generated method stub
		int identifier = id.addAndGet(1);
		byte[] sendMsg = PublishMsgBuilder.publish(identifier, false, 0, false, topic, txt.getBytes(),
				txt.getBytes().length);
		sendBytes(sendMsg);
	}

	private void updateRcvPublish(String topic, String rcv) {
		father.updateRcvPublish(topic, rcv);
	}

	@Override
	public void run() {
		int MAXSIZE = 81920;
		byte[] readbuf = new byte[MAXSIZE];
		// int length=0;
		while (dorecv.get()) {
			if (connected.get() && socket != null && mqttInputStream != null) {
				try {
					//
					int length = mqttInputStream.readBytes(readbuf);
					if (length == 0) {
						continue;
					}
					if(length<0) {
						OnDisconnect();
					}
					MqttData mqttData =MqttParser.parse(readbuf,length);
					father.updateRcvCommand(mqttData.getCmdType().name());
					if (mqttData.getCmdType() == CmdType.PUBLISH) {
						PublishMqttData p = (PublishMqttData) mqttData;
						String payld = new String(p.getPayload());
						updateRcvPublish(p.getTopic(), payld);
						father.printInfo("收到消息：" + p.getTopic());
						PublishMqttData publishMqttData=(PublishMqttData)mqttData;
						if(publishMqttData.getQos()==1) {
							puback(publishMqttData.getPackageId());
						}else if(publishMqttData.getQos()==2) {
							//PUBLISH(3),PUBACK(4),
							pubrec(publishMqttData.getPackageId());
							//PUBREC(5),PUBREL(6),PUBCOMP(7)
						}
					} else if (mqttData.getCmdType() == CmdType.PINGREQ) {
						pong();
					}else if(mqttData.getCmdType() == CmdType.PUBREC){
						pubrel(((PubRecMqttData)mqttData).getPackageId());
					}else if(mqttData.getCmdType() == CmdType.PUBREL) {
						pubcomp(((PubRelMqttData)mqttData).getPackageId());
					} else if(mqttData.getCmdType() == CmdType.NIL) {
						updateRcvTime();
						continue;
					}else if(mqttData.getCmdType()==CmdType.DISCONNECT) {
						OnDisconnect();
					} else {
						father.printInfo("收到其他消息：" + mqttData.getCmdType());
					}
					updateRcvTime();
				} catch (SocketTimeoutException e) {
					checkPing();
				} catch (Throwable e) {
					e.printStackTrace();
					father.printInfo(e.toString());
				}
			}
		}
	}

	private void OnDisconnect() {
		// TODO Auto-generated method stub
		dorecv.set(false);
		connected.set(false);
		father.printInfo("连接被关闭");
		father.connectFail();
		if (socket != null) {
			try {
				socket.close();
			} catch (IOException e) {
			}
		}
		if (executorService != null) {
			executorService.shutdown();
		}
	}

	private void updateRcvTime() {
		lastRcvTime = System.currentTimeMillis();
	}

	public void checkPing() {
		if (keepLive <= 0) {
			return;
		}
		long timespan = System.currentTimeMillis() - lastRcvTime;
		if (keepLive * 1000 <= timespan) {
			try {
				lastRcvTime = System.currentTimeMillis();
				ping();
			} catch (IOException e) {
				//e.printStackTrace();
			}
		}
	}
}
